package com.example.kafka.service;

import com.example.kafka.mock.MockData;
import com.example.kafka.util.KafkaRetryUtils;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.RetriableException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * Kafka异步消息发送Service
 */
@Service
public class KafkaAsyncService {

    @Autowired
    private Producer<String, String> producer;

    /**
     * 用例1：标准异步发送+回调
     */
    public void sendAsyncStandard() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                // 记录异常
            } else {
                // 处理成功
            }
        });
    }

    /**
     * 用例2：异步发送+回调重试
     */
    public void sendAsyncWithRetry() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        sendWithRetry(record, 0, 3);
    }

    private void sendWithRetry(ProducerRecord<String, String> record, int retryCount, int maxRetries) {
        producer.send(record, (metadata, exception) -> {
            if (exception != null && KafkaRetryUtils.isRetriable(exception)) {
                if (retryCount < maxRetries) {
                    sendWithRetry(record, retryCount + 1, maxRetries);
                } else {
                    // 重试耗尽处理
                }
            } else if (exception != null) {
                // 不可重试异常处理
            } else {
                // 发送成功
            }
        });
    }

    /**
     * 用例3：异步发送+回调异常类型细分
     */
    public void sendAsyncWithErrorType() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                if (exception instanceof org.apache.kafka.common.errors.AuthorizationException) {
                    // 鉴权异常处理
                } else if (KafkaRetryUtils.isRetriable(exception)) {
                    // 可重试异常处理
                } else {
                    // 其他异常
                }
            } else {
                // 发送成功
            }
        });
    }

    /**
     * 用例4：异步发送+回调补偿
     */
    public void sendAsyncWithCompensation() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        producer.send(record, (metadata, exception) -> {
            if (exception != null && !(exception instanceof RetriableException)) {
                // 补偿处理
                compensate(record, exception.getMessage());
            }
        });
    }

    /**
     * 用例5：异步发送+最大重试次数与退避
     */
    public void sendAsyncWithMaxRetryAndBackoff() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        AtomicInteger retryCount = new AtomicInteger(0);
        int maxRetries = 3;
        sendWithBackoff(record, retryCount, maxRetries, 200);
    }

    private void sendWithBackoff(ProducerRecord<String, String> record, AtomicInteger retryCount, int maxRetries, long backoffMs) {
        producer.send(record, (metadata, exception) -> {
            if (exception != null && KafkaRetryUtils.isRetriable(exception)) {
                if (retryCount.incrementAndGet() <= maxRetries) {
                    try {
                        Thread.sleep(backoffMs * retryCount.get());
                    } catch (InterruptedException ignored) {}
                    sendWithBackoff(record, retryCount, maxRetries, backoffMs);
                } else {
                    // 重试耗尽处理
                }
            } else if (exception != null) {
                // 不可重试异常处理
            } else {
                // 发送成功
            }
        });
    }

    /**
     * 补偿逻辑（示例）
     */
    private void compensate(ProducerRecord<String, String> record, String reason) {
        // TODO: 实现补偿逻辑，如写入本地消息表
    }
}